package com.apex.spark.stream;

import com.apex.env.Execution;
import com.apex.spark.SparkEnvironment;
import com.apex.spark.transform.SparkTransform;
import com.typesafe.config.Config;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.streaming.StreamingContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * spark 流处理环境入口
 */
public class SparkStreamingExecution implements Execution<SparkStreamingSource, SparkTransform
        ,SparkStreamingSink> {
    //配置文件管理
    private Config config;
    private final SparkEnvironment sparkEnvironment;

    /**
     * 初始化spark流处理环境
     * @param sparkEnvironment 上下文执行环境
     */
    public SparkStreamingExecution(SparkEnvironment sparkEnvironment) {
        this.sparkEnvironment = sparkEnvironment;
        sparkEnvironment.createStreamingContext();
    }

    /**
     * 数据处理方法
     * @param sources 数据源
     * @param transforms 业务处理
     * @param sinks 写入目标数据
     */
    @Override
    public void start(List<SparkStreamingSource> sources, List<SparkTransform> transforms,
                      List<SparkStreamingSink> sinks) {
        String tableName = sparkEnvironment.getQueryTable();
        List<Dataset<Row>> datasets = new ArrayList<>();

        Dataset<Row> result = null;

        if (!sources.isEmpty()){
            for (SparkStreamingSource source : sources){
                Dataset<Row> dataset = source.getData(sparkEnvironment);
                datasets.add(dataset);
                registerTempView(tableName,dataset);
            }

            Dataset<Row> dataset = datasets.get(0);

            for (SparkTransform transform : transforms){
              Dataset<Row> dataset1 =  transform.process(sparkEnvironment,dataset);
              if (Objects.isNull(dataset1)){
                  result = dataset;
              }
            }

            for (SparkStreamingSink sink : sinks){
                sink.outputStreaming(sparkEnvironment,result);
            }
        }
        StreamingContext streamingContext = sparkEnvironment.getStreamingContext();
        streamingContext.start();
        streamingContext.awaitTermination();
    }

    /**
     * 程序执行状态
     * @return boolean
     */
    @Override
    public boolean status() {
        return false;
    }

    /**
     *
     * @param plugin 执行环境
     */
    @Override
    public void prepare(Void plugin) {

    }

    /**
     *
     * @return config
     */
    @Override
    public Config getConfig() {
        return config;
    }

    /**
     *
     * @param config config
     */
    @Override
    public void setConfig(Config config) {
        this.config = config;
    }

    /**
     * 注册临时表结果集
     * @param tableName 表名称
     * @param dataSet 查询结果集
     */
    private void registerTempView(String tableName, Dataset<Row> dataSet) {
        dataSet.createOrReplaceTempView(tableName);
    }
}
